from confluent_kafka import Producer
from django.http import JsonResponse


def delivery_report(err, msg):
    """Callback for message delivery status."""
    if err is not None:
        print(f"Message delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}]")


def sendMessage(request):
    # SASL/PLAIN配置
    config = {
        'bootstrap.servers': '192.168.88.8:9092,192.168.88.7:9092,192.168.88.6:9092',  # 替换为你的Kafka服务器地址
        'security.protocol': 'SASL_PLAINTEXT',
        'sasl.mechanisms': 'PLAIN',
        'sasl.username': 'user2',  # 替换为你的Kafka用户名
        'sasl.password': '123456',  # 替换为你的Kafka密码
    }

    # 创建生产者实例
    producer = Producer(config)

    # 发送消息
    for i in range(10):
        message = f"Message #{i}"
        producer.produce(
            'hainiu',  # 替换为你的主题名称
            key=str(i),  # 可选：消息键
            value=message.encode('utf-8'),  # 将字符串转换为字节串，Kafka原生处理字节数据
            callback=delivery_report,
        )

    # 确保所有消息都已发送
    producer.flush()
    xiao_ming = {"name": "xiaoming", "age": 18}
    return JsonResponse(xiao_ming);


